package com.client.zk;


import com.common.loadbalance.CommonLoadBalance;
import com.common.tools.CommonTool;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.listen.ListenerContainer;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.RetryOneTime;

import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author Zixiang Hu
 * @description
 * @create 2021-03-04-16:57
 */
@Slf4j
public class ZkService {
    private CuratorFramework client;

    // 负载均衡算法
    private CommonLoadBalance loadBalance;
    // 本地缓存的服务提供的列表
    // 键是服务名，值是ServiceInfo，包括服务的获取事件和提供该服务的IP列表
    private ConcurrentHashMap<String, ServiceInfo> serviceCache = new ConcurrentHashMap<>();

    class ServiceInfo {
        long time;
        List<String> serviceList;
    }

    public ZkService(CuratorFramework client, String connectString, int code) {
        this(connectString, code);
        this.client = client;
    }


    public ZkService(String connectString, int code) {
        this.loadBalance = CommonLoadBalance.getLoadBalanceByCode(code);
        client = CuratorFrameworkFactory.builder().connectString(connectString)
                // 会话超时时间
                .sessionTimeoutMs(5000)
                // 超时重连策略
                .retryPolicy(new RetryOneTime(1000))
                // 配置默认的工作空间
                .namespace("myrpc")
                .build();
        client.start();
        log.info("连接Zookeeper服务器成功，服务器地址为：{}", connectString);
    }


    /**
     * 服务发现功能
     *
     * @param serviceName 服务名称
     * @return
     */
    public InetSocketAddress discoveryService(String serviceName) {
        try {
            // 2021.05.08 新增
            List<String> list = needUpdate(serviceName);
            // 如果本地缓存中不能够获得到连接列表，则通过连接zk获取
            if (list == null || list.size() == 0) {
                list = client.getChildren().forPath("/" + serviceName);
                updateCache(serviceName, list);
            } else {
                // 如果本地缓存中可以获取到连接
                // 使用负载均衡算法
                String addr = loadBalance.getServerAddr(list);
                String[] split = addr.split(":");
                InetSocketAddress socketAddress = new InetSocketAddress(split[0], Integer.parseInt(split[1]));
                // 判断本地连接中的服务可否连接成功，如果失败则重新从zk中获取
                if (CommonTool.canConnect(socketAddress))
                    return socketAddress;
                else {
                    list = client.getChildren().forPath("/" + serviceName);
                    updateCache(serviceName, list);
                }
            }
            /*
             * 什么情况下会走下面的if判断？
             * 1.当本地缓存中没有缓存当前提供的服务或者服务已经过期，此时需要重新从ZK中获取
             * 2.本地缓存中有当前服务，但是服务提供的IP已经不能通信，则需要重新从ZK中获取
             * */
            if (list != null && list.size() != 0) {
                // 使用负载均衡算法
                String addr = loadBalance.getServerAddr(list);
                String[] split = addr.split(":");
                InetSocketAddress socketAddress = new InetSocketAddress(split[0], Integer.parseInt(split[1]));
                return socketAddress;
            } else {
                log.error("{}服务没有对应的服务器提供", serviceName);
            }
        } catch (Exception e) {
            client.close();
            log.error("ZkService的discoveryService发生异常，客户端被迫关闭，异常信息是：{}", e.getMessage());
        }
        return null;
    }

    /**
     * 判断当前本地缓存的服务是否需要更新，如果超过两小时则进行一个更新
     *
     * @param serviceName
     * @return 如果返回空则表示需要连接zk进行服务更新，否则返回本地缓存的服务列表
     */
    public List<String> needUpdate(String serviceName) {
        if (serviceCache.containsKey(serviceName)) {
            ServiceInfo serviceInfo = serviceCache.get(serviceName);
            long nowTime = System.currentTimeMillis();
            long diff = nowTime - serviceInfo.time;
            // 如果大于2小时，则进行更新
            if (diff / 1000 > 60 * 60 * 2) {
                // 删除过期缓存
                serviceCache.remove(serviceName);
                return null;
            } else
                return serviceInfo.serviceList;
        }
        return null;
    }

    /**
     * 对本地缓存提供的服务进行更新
     *
     * @param serviceName 需要更新的服务名
     * @param list        提供该服务的服务器IP地址
     */
    public void updateCache(String serviceName, List<String> list) {
        ServiceInfo serviceInfo = new ServiceInfo();
        serviceInfo.time = System.currentTimeMillis();
        serviceInfo.serviceList = list;
        serviceCache.put(serviceName, serviceInfo);
    }

    /**
     * 监听机制，暂时没有使用
     public void zkWatcher(String serviceName) {
     PathChildrenCache pathChildrenCache = new PathChildrenCache(client, "/" + serviceName, true);
     try {
     pathChildrenCache.start();
     ListenerContainer<PathChildrenCacheListener> listenable = pathChildrenCache.getListenable();
     listenable.addListener((curatorFramework, pathChildrenCacheEvent) -> {
     System.err.println("type = " + pathChildrenCacheEvent.getType());
     System.err.println("path = " + pathChildrenCacheEvent.getData().getPath());
     System.err.println("data = " + new String(pathChildrenCacheEvent.getData().getData()));
     });
     } catch (Exception e) {
     e.printStackTrace();
     }
     }
     */
}
